from flask import Flask, render_template
from flask_socketio import SocketIO
from flask_cors import CORS
from kafka import KafkaConsumer
import json

app = Flask(__name__, template_folder='templates')
app.config['SECRET_KEY'] ='secret!'
socketio = SocketIO(app)

# 配置不同topic对应的Kafka消费者
def create_kafka_consumer(topic, group_id):
    return KafkaConsumer(
        topic,
        auto_offset_reset='earliest',
        bootstrap_servers="192.168.121.129:9092",
        group_id=group_id,
        value_deserializer=lambda x: json.loads(x.decode('utf-8'))
    )

consumer1 = create_kafka_consumer('data1', 'y1')
consumer2 = create_kafka_consumer('data2', 'y2')
consumer3 = create_kafka_consumer('data3', 'y3')
consumer4 = create_kafka_consumer('data4', 'y4')
consumer5 = create_kafka_consumer('data5', 'y5')

# 定义函数用于读取并推送data1的Kafka数据
def push_data1():
    for msg in consumer1:
        socketio.emit('data1', msg.value)

# 定义函数用于读取并推送data2的Kafka数据
def push_data2():
    for msg in consumer2:
        socketio.emit('data2', msg.value)

# 定义函数用于读取并推送data3的Kafka数据
def push_data3():
    for msg in consumer3:
        socketio.emit('data3', msg.value)

# 定义函数用于读取并推送data4的Kafka数据
def push_data4():
    for msg in consumer4:
        socketio.emit('data4', msg.value)

# 定义函数用于读取并推送data5的Kafka数据
def push_data5():
    for msg in consumer5:
        socketio.emit('data5', msg.value)

# 启动后台任务来推送各个topic的数据
@socketio.on('connect')
def handle_connect():
    socketio.start_background_task(push_data1)
    socketio.start_background_task(push_data2)
    socketio.start_background_task(push_data3)
    socketio.start_background_task(push_data4)
    socketio.start_background_task(push_data5)

if __name__ == '__main__':
    socketio.run(app, host='0.0.0.0', port=4444, debug=True)